package com.idroc.message.impl;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.mina.common.IdleStatus;
import org.apache.mina.common.IoHandler;
import org.apache.mina.common.IoSession;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.transport.socket.nio.SocketAcceptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.idroc.message.Message;
import com.idroc.message.MessageHandler;
import com.idroc.message.MessageService;
import com.idroc.message.MessageType;
import com.idroc.message.filter.codec.MessageCodecFactory;

/**
 * 消息服务器
 * @author 赤脚詹(yuyoo4j@163.com)
 * @date Aug 17, 2009 4:14:54 PM 
 * @memo:实现MessageService接口.作为消息通讯的服务端.底层通讯依赖mina功能
 */
public class MessageServer implements MessageService {
 
	private static final Logger log = LoggerFactory.getLogger(MessageServer.class);
	
	/**
	 * 提供服务主体的属性名称
	 */
	public static final String MSG_SERVICE_HOST = "msg_service_host";
	
	/**
	 * 监听端口属性名称
	 */
	public static final String MSG_SERVICE_PORT = "msg_service_port";
	
	/**
	 * 消息处理类容器
	 */
	private final Map<Class<? extends Message>, MessageHandler> handlers = new ConcurrentHashMap<Class<? extends Message>, MessageHandler>();
	/**
	 * 配置信息
	 */
	private final Map<String, String> config = new HashMap<String, String>();
	/**
	 * 是否被扣住
	 */
	private boolean pinned = false;
	/**
	 * 是否已启动
	 */
	private boolean started = false;
	/**
	 * 是否已停止
	 */
	private boolean stopped = false;
	/**
	 * mina SocketAcceptor
	 */
	private SocketAcceptor acceptor = null;
	/**
	 * mina IoHandler
	 */
	private IoHandler handler = null; 
		
	/**
	 * 根据注册地消息类对象查找某个消息处理器
	 */
	public MessageHandler findMsgHandler(Class<? extends Message> msgClazz) {
		return handlers.get(msgClazz);
	}

	/**
	 * 是否存在某个消息处理器
	 */
	public boolean hasMsgHandler(MessageHandler handler) {		
		return handlers.containsValue(handler);
	}

	/**
	 * 初始化
	 */
	public void init(Map<String, String> config) {

		this.config.putAll(config);
	}

	/**
	 * 是否被扣住
	 */
	public boolean isPinned() { 
		return pinned;
	}

	public boolean isStarted() { 
		return started;
	}

	/**
	 * 是否停止
	 */
	public boolean isStopped() { 
		return stopped;
	}

	/**
	 * 扣住动作,消息服务器不会在发送各种消息事件.
	 */
	public void pin() {
		pinned = true;
	}
	
	/**
	 * 解除扣住动作
	 */
	public void unpin() {
		pinned = false;
	}

	/*
	 * (non-Javadoc)
	 * @see com.idroc.mina.multiplexed.message.MessageService#registerMsgHandler(java.lang.Class, com.idroc.mina.multiplexed.message.MessageHandler)
	 */
	public void registerMsgHandler(Class<? extends Message> msgClazz,
			MessageHandler handler) {
		
		if (null != msgClazz && null != handler) {
			MessageType.registry(msgClazz);
			handlers.put(msgClazz, handler);
			handler.onAttached(this); // 发生注册事件
		} else {
			String msg = "非法参数,两个参数都不能为null. msgClazz:" + msgClazz + ", handler:" + handler;
			log.error(msg);
			throw new IllegalArgumentException(msg);
		}
	} 

	/*
	 * (non-Javadoc)
	 * @see com.idroc.mina.multiplexed.message.MessageService#start()
	 */
	public void start() {
		
		if (!started) {
			
			String port = config.get(MSG_SERVICE_PORT);
			
			handler = new MsgIoHandler();
			acceptor = new SocketAcceptor();
			acceptor.getFilterChain().addLast("protocol", new ProtocolCodecFilter(new MessageCodecFactory()));
			
			String host = config.get(MSG_SERVICE_HOST); 
			try {
				if (null != host) {
					acceptor.bind(new InetSocketAddress(host, Integer.parseInt(port)), handler);
				} else {
					acceptor.bind(new InetSocketAddress(Integer.parseInt(port)), handler);
				}
			} catch (IOException ex) { 
				String errorMsg = "启动消息服务器异常";
				log.error(errorMsg, ex);
				throw new RuntimeException(errorMsg, ex);
			}
			
			stopped = false;
			started = true;
			
			log.info("服务器启动成功. 监听端口:{}", port);
		}
	}

	/*
	 * (non-Javadoc)
	 * @see com.idroc.mina.multiplexed.message.MessageService#stop()
	 */
	public void stop() {

		if (started && !stopped) {
			acceptor.unbindAll();
			started = false;
			stopped = true;
		}
	}

	/*
	 * (non-Javadoc)
	 * @see com.idroc.mina.multiplexed.message.MessageService#unregisterMsgHandler(java.lang.Class)
	 */
	public MessageHandler unregisterMsgHandler(Class<? extends Message> msgClazz) {
		
		MessageHandler handler = handlers.remove(msgClazz);
		if (null != handler) {
			handler.onReleased(this); // 发生释放事件
		}
		return handler;
	}
	
	/**
	 * 消息到达
	 * @param session -- 传递消息的会话
	 * @param message -- 消息
	 * <p>
	 * 说明:<br>  
	 * </p>
	 */
	protected void onMsgReceived(IoSession session, Message message) {
		
		if (null != message) {
			MessageHandler handler = handlers.get(message.getClass());
			if (null != handler) {
				handler.onMsgReceived(session, message); // 发生消息到达事件
			}
		}
	}
	
	/**
	 * 消息发送
	 * @param session -- 传递消息的会话
	 * @param message -- 消息
	 * <p>
	 * 说明:<br>  
	 * </p>
	 */
	protected void onMsgSend(IoSession session, Message message) {

		if (null != message) {
			MessageHandler handler = handlers.get(message.getClass());
			if (null != handler) {
				handler.onMsgSent(session, message); // 发生消息发送事件
			}
		}
	}  
	
	/**
	 * mina IoHandler接口的实现
	 * @author 詹都林(zhandl@tianya.cn) <br>
	 * @date May 23, 2009 12:01:55 AM <br>
	 * @version 1.0.0
	 * @descript: 类功能简称<br>
	 *            <p>
	 *            备注:<br>
	 *            </p>
	 */
	private class MsgIoHandler implements IoHandler {

		public void exceptionCaught(IoSession session, Throwable ex)
				throws Exception { 
			
			log.error("MsgIoHandler exceptionCaught. IoSession:{}, Throwable:{}", session, ex);
			throw (Exception)ex;
		}

		public void messageReceived(IoSession session, Object obj)
				throws Exception {
						
			if (obj instanceof Message) {
				Message message = (Message)obj; 
				MessageServer.this.onMsgReceived(session, message);
			} else {
				log.info("非消息对象到达. Object:" + obj);
			}
		}

		public void messageSent(IoSession session, Object obj) throws Exception {
						
			if (obj instanceof Message) { 
				Message message = (Message)obj;
				MessageServer.this.onMsgSend(session, message);
			} else {
				log.info("非消息对象发送. Object:" + obj);
			}
		}

		public void sessionClosed(IoSession session) throws Exception {
			
			log.debug("MsgIoHandler sessionClosed. session:" + session);
		}

		public void sessionCreated(IoSession session) throws Exception {
			
			log.debug("MsgIoHandler sessionCreated. session:" + session);
		}

		public void sessionIdle(IoSession session, IdleStatus status)
				throws Exception {			
			
			log.debug("MsgIoHandler sessionIdle. IoSession:{}, IdleStatus:{}", session, status);
		}

		public void sessionOpened(IoSession session) throws Exception {
			
			log.debug("MsgIoHandler sessionOpened. session:" + session);
		}
		
	}

}
